package com.shujia.core.transformations

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/*
  结论：
    1、默认情况针对同一个DAG有向无环图，若执行多次，从第二次开始，spark会将宽依赖之前的数据进行缓存
    2、若上面的多次执行，只要其中一个rdd的计算逻辑不一样，依旧是从第一个rdd开始计算


  缓存：避免RDD重复计算
    cache指的是将多个job作业相同的rdd进行缓存
 */
object cacheOpt {
  def main(args: Array[String]): Unit = {
    //计算每个班级的人数
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("cache合并")

    val sc = new SparkContext(conf)
    //================================================================
    val studentRDD: RDD[Student] = sc.textFile("spark/data/students.txt")
      .map(_.split(","))
      .map {
        case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
          Student(id.toLong, name, age.toInt, gender, clazz)
      }

    /**
     * 针对多个job作业共同使用的RDD进行缓存
     * 当某一个RDD其中的计算逻辑非常复杂耗时的时候，而且这个RDD会被多个job作业复用
     * 就可以将这个RDD的计算逻辑以及计算结果进行缓存
     */
    //    studentRDD.cache()
    //使用persist()函数对RDD进行缓存，并设置缓存级别
    studentRDD.persist(StorageLevel.DISK_ONLY)


    //计算每个班级的人数
    val resRDD1: RDD[(String, Int)] = studentRDD.map((stu: Student) => (stu.clazz, 1))
      .reduceByKey(_ + _)

    //计算每种年龄的人数
    val resRDD2: RDD[(Int, Int)] = studentRDD.map((stu: Student) => (stu.age, 1))
      .reduceByKey(_ + _)

    resRDD1.foreach(println)
    resRDD2.foreach(println)

    while (true) {

    }
  }
}

case class Student(id: Long, name: String, age: Int, gender: String, clazz: String)
